package com.chis.jmdatatimer.yfjz;


import com.chis.jmdatatimer.comm.PropertyConfigurer;

import java.util.Properties;

/**
 * @author wlj
 * @version 2019年01月10日
 * @Description
 * @Company zwx
 */
public abstract class KafkaConsumerTemplate {

    protected Properties initKafkaData() {
        Properties props = new Properties();
        props.put("bootstrap.servers", PropertyConfigurer.getProperty("kafka.servers"));
        props.put("group.id", getGroupId());
        props.put("enable.auto.commit", "true");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "com.chis.jmdatatimer.comm.DecodeingKafka");
        props.put("partitioner.class", "org.apache.kafka.clients.producer.internals.DefaultPartitioner");
        /* 自动确认offset的时间间隔  */
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        //消息发送的最长等待时间.需大于session.timeout.ms这个时间
        props.put("request.timeout.ms", "40000");
        //一次从kafka中poll出来的数据条数
        //max.poll.records条数据需要在在session.timeout.ms这个时间内处理完
        props.put("max.poll.records", "10");
        //server发送到消费端的最小数据，若是不满足这个数值则会等待直到满足指定大小。默认为1表示立即接收。
        props.put("fetch.min.bytes", "1");
        //若是不满足fetch.min.bytes时，等待消费端请求的最长等待时间
        props.put("fetch.wait.max.ms", "1000");

        return props;
    }

    public abstract String getGroupId();

    public abstract String getTopic();

    public abstract void run();


}
